package zhouhong;

import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * Create by zhh on 2022/8/30.
 */
public class WC {
    public static void main(String[] args) throws Exception{
        //创建上下文
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //对接数据源
        DataStreamSource<String> source = env.socketTextStream("192.168.32.222", 9527);
        //业务逻辑处理
        source.flatMap(new ZHflatMap())//获取分割后的数据
                .filter(new ZHFilter())//去掉空值
                .map(new ZHMap())//转成（value，1）这样的二元组
                .keyBy(new ZHKeyBy())//按照第一位聚合元组
                .sum(1)//求和第二位置
                .print();
        env.execute("Test");//执行（流使用）
    }
}

class ZHflatMap implements FlatMapFunction<String, String>{
    @Override
    public void flatMap(String value, Collector<String> out) throws Exception {
        String[] words = value.split(",");
        for (String word : words){
            out.collect(word.toLowerCase().trim());//转小写，去空格
        }
    }
}

class ZHFilter implements FilterFunction<String> {
    @Override
    public boolean filter(String value) throws Exception {
        return StringUtils.isNotEmpty(value);//获取不为空
    }
}

class ZHMap implements MapFunction<String, Tuple2<String,Long>>{
    @Override
    public Tuple2<String, Long> map(String value) throws Exception {
        return new Tuple2<>(value,1L);
    }
}

class ZHKeyBy implements KeySelector<Tuple2<String, Long>, String>{
    @Override
    public String getKey(Tuple2<String, Long> value) throws Exception {
        return value.f0;//获取第0位数据
    }
}